package com.hdw.job.batch.writer;

import com.hdw.common.base.BaseException;
import com.hdw.common.db.DynamicDataSource;
import com.hdw.common.db.HikariDataSourceExt;
import com.hdw.common.util.DBUtil;
import com.hdw.common.util.EtlUtil;
import com.hdw.job.batch.element.Column;
import org.apache.commons.lang3.tuple.Triple;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.item.ItemWriter;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.sql.Connection;
import java.sql.PreparedStatement;
import java.sql.SQLException;
import java.sql.Types;
import java.util.*;

@Component
@StepScope
public class JDBCWriter implements ItemWriter<Map> {

    @Value("#{jobParameters[idDbTar]}")
    String idDbTar;
    @Value("#{jobParameters[tableName]}")
    String tbTar;


    private HikariDataSourceExt dwDataSource;
    private List<String> tarColumns;
    private List<String> queryColumns;
    // ColumnName -  Left:ColumnName Middle:CuolmnType Right:ColumnTypeName
    private Map<String, Triple<String, Integer, String>> outTableMetaData;
    Connection connection = null;
    PreparedStatement preparedStatement = null;

    public JDBCWriter() {
    }

    public JDBCWriter(String idDbTar, String tbTar) {
        this.idDbTar = idDbTar;
        this.tbTar = tbTar;
    }

    @Override
    public void write(List<? extends Map> list) throws Exception {
        try {
            if (dwDataSource == null) dwDataSource = DynamicDataSource.getHikariDataSourceById(idDbTar);
            if (dwDataSource == null) throw new BaseException(String.format("数据源获取失败！"));
            if (connection == null) connection = dwDataSource.getConnection();
            if (outTableMetaData == null) outTableMetaData = DBUtil.getColumnMetaDataToMap(connection, tbTar, " * ");

            //获取可以写入的列
            if (tarColumns == null || queryColumns == null)
                enableColumns(outTableMetaData.keySet(), list.get(0).keySet());

            connection.setAutoCommit(false);
            if (preparedStatement == null) preparedStatement = connection.prepareStatement(buildSql());

            for (Map map : list) {
                preparedStatement = fillPreparedStatement(preparedStatement, map);
                preparedStatement.addBatch();
            }
            preparedStatement.executeBatch();
            connection.commit();
        } catch (Exception e) {
            connection.rollback();
            throw new BaseException("写入时错误:" + e.getMessage());
        }
    }

    /**
     * 获取可以使用的列名
     * 查询SQL列名与输入表列名一致,忽略大小写
     *
     * @param tarTableColumns
     * @param sqlQueryColumns
     * @return
     */
    protected void enableColumns(Set<String> tarTableColumns, Set sqlQueryColumns) {
        List<String> queryColumns = new ArrayList<>();
        List<String> tarColumns = new ArrayList<>();

        Map<String, String> queryColumnsUpperMap = new HashMap();
        Map<String, String> tarColumnsUpperMap = new HashMap();
        for (Object sqlQueryColumn : sqlQueryColumns) {
            queryColumnsUpperMap.put(sqlQueryColumn.toString().toUpperCase(), sqlQueryColumn.toString());
        }
        for (String tarTableColumn : tarTableColumns) {
            tarColumnsUpperMap.put(tarTableColumn.toUpperCase(), tarTableColumn);
        }

        for (String tarTableCol : tarColumnsUpperMap.keySet()) {
            if (queryColumnsUpperMap.containsKey(tarTableCol)) {
                queryColumns.add(queryColumnsUpperMap.get(tarTableCol));
                tarColumns.add(tarColumnsUpperMap.get(tarTableCol));
            }
        }

        this.queryColumns = queryColumns;
        this.tarColumns = tarColumns;

    }

    // 直接使用了两个类变量：columnNumber,outTableMetaData
    protected PreparedStatement fillPreparedStatement(PreparedStatement preparedStatement, Map record)
            throws SQLException, BaseException {

        for (int i = 0; i < tarColumns.size(); i++) {
            int columnSqltype = this.outTableMetaData.get(tarColumns.get(i)).getMiddle();
            String typeName = this.outTableMetaData.get(tarColumns.get(i)).getRight();
            String s = queryColumns.get(i);
            Object o = record.get(s);
            Column column = null;
            if (o instanceof Column) {
                column = (Column) o;
            }
            preparedStatement = fillPreparedStatementColumnType(preparedStatement, i, columnSqltype, typeName, column);
        }

        return preparedStatement;
    }

    protected PreparedStatement fillPreparedStatementColumnType(PreparedStatement preparedStatement, int columnIndex,
                                                                int columnSqltype, String typeName, Column column) throws SQLException, BaseException {
        Date utilDate;
        switch (columnSqltype) {
            case Types.CHAR:
            case Types.NCHAR:
            case Types.CLOB:
            case Types.NCLOB:
            case Types.VARCHAR:
            case Types.LONGVARCHAR:
            case Types.NVARCHAR:
            case Types.LONGNVARCHAR:
                preparedStatement.setString(columnIndex + 1, column.asString());
                break;

            case Types.SMALLINT:
            case Types.INTEGER:
            case Types.BIGINT:
            case Types.NUMERIC:
            case Types.DECIMAL:
            case Types.FLOAT:
            case Types.REAL:
            case Types.DOUBLE:
                String strValue = column.asString();
                if ("".equals(strValue)) {
                    preparedStatement.setString(columnIndex + 1, null);
                } else {
                    preparedStatement.setString(columnIndex + 1, strValue);
                }
                break;

            //tinyint is a little special in some database like mysql {boolean->tinyint(1)}
            case Types.TINYINT:
                Long longValue = column.asLong();
                if (null == longValue) {
                    preparedStatement.setString(columnIndex + 1, null);
                } else {
                    preparedStatement.setString(columnIndex + 1, longValue.toString());
                }
                break;

            // for mysql bug, see http://bugs.mysql.com/bug.php?id=35115
            case Types.DATE:
                if (typeName == null) {
                    typeName = this.outTableMetaData.get(tarColumns.get(columnIndex)).getRight();
                }

                if (typeName.equalsIgnoreCase("year")) {
                    if (column.asBigInteger() == null) {
                        preparedStatement.setString(columnIndex + 1, null);
                    } else {
                        preparedStatement.setInt(columnIndex + 1, column.asBigInteger().intValue());
                    }
                } else {
                    java.sql.Date sqlDate = null;
                    try {
                        utilDate = column.asDate();
                    } catch (BaseException e) {
                        throw new SQLException(String.format("Date 类型转换错误：[%s]", column));
                    }

                    if (null != utilDate) {
                        sqlDate = new java.sql.Date(utilDate.getTime());
                    }
                    preparedStatement.setDate(columnIndex + 1, sqlDate);
                }
                break;

            case Types.TIME:
                java.sql.Time sqlTime = null;
                try {
                    utilDate = column.asDate();
                } catch (BaseException e) {
                    throw new SQLException(String.format(
                            "TIME 类型转换错误：[%s]", column));
                }

                if (null != utilDate) {
                    sqlTime = new java.sql.Time(utilDate.getTime());
                }
                preparedStatement.setTime(columnIndex + 1, sqlTime);
                break;

            case Types.TIMESTAMP:
                java.sql.Timestamp sqlTimestamp = null;
                try {
                    utilDate = column.asDate();
                } catch (BaseException e) {
                    throw new SQLException(String.format("TIMESTAMP 类型转换错误：[%s]", column));
                }

                if (null != utilDate) {
                    sqlTimestamp = new java.sql.Timestamp(
                            utilDate.getTime());
                }
                preparedStatement.setTimestamp(columnIndex + 1, sqlTimestamp);
                break;

            case Types.BINARY:
            case Types.VARBINARY:
            case Types.BLOB:
            case Types.LONGVARBINARY:
                preparedStatement.setBytes(columnIndex + 1, column
                        .asBytes());
                break;

            case Types.BOOLEAN:
                preparedStatement.setString(columnIndex + 1, column.asString());
                break;

            // warn: bit(1) -> Types.BIT 可使用setBoolean
            // warn: bit(>1) -> Types.VARBINARY 可使用setBytes
            case Types.BIT:
                if (dwDataSource.getEuTp() == DBUtil.DB_TYPE_MYSQL) {
                    preparedStatement.setBoolean(columnIndex + 1, column.asBoolean());
                } else {
                    preparedStatement.setString(columnIndex + 1, column.asString());
                }
                break;
            default:
                throw new BaseException(String.format(
                        "不支持数据库写入这种字段类型. 字段名:[%s], 字段类型:[%d], 字段Java类型:[%s]. 请修改表中该字段的类型或者不同步该字段.",
                        this.outTableMetaData.get(tarColumns.get(columnIndex)).getLeft(),
                        this.outTableMetaData.get(tarColumns.get(columnIndex)).getMiddle(),
                        this.outTableMetaData.get(tarColumns.get(columnIndex)).getRight()));
        }
        return preparedStatement;
    }

    protected String buildSql() {
        String insertSql = EtlUtil.getInsertSQL(tarColumns, tbTar);
        return insertSql;
    }


    public void close() {
        DBUtil.closeDBResources(null, preparedStatement, connection);
    }

    @Override
    protected void finalize() throws Throwable {
        this.close();
        super.finalize();
    }
}
